use model_graph_types::channel::pubsub::{Topic, TopicMessage};
use std::{
    collections::HashMap,
    sync::{Arc, Mutex, RwLock},
};

use crate::pubsub::topic_exist;

use super::topic_subscribed;

pub async fn on_data_channel_message(peer_id: &String, data: Vec<u8>) -> anyhow::Result<()> {
    let mut message: TopicMessage = bincode::deserialize::<TopicMessage>(&data[..])
        .map_err(|err| anyhow::anyhow!("{}", err))?;

    let topic = &message.topic;

    if !topic_subscribed(topic)? {
        //接收到了没有订阅的消息
        //查看是否是拥有的主题
        if !topic_exist(topic)? {
            tracing::debug!("没有订阅该主题:{}", topic);
            return Ok(());
        }
    }
    message.src = Some(peer_id.clone());

    //调用消息回调
    tracing::debug!("on_data_channel_message:{}", topic);
    if let Ok(callbacks) = get_on_message_callback(topic) {
        tracing::debug!("callbacks.....",);
        callbacks
            .clone()
            .lock()
            .map_err(|err| anyhow::anyhow!("{}", err))?
            .iter()
            .all(|callback| {
                tracing::debug!("message callback.....",);
                callback(message.clone());
                true
            });
    }
    //调用消息回调
    tracing::debug!("on_data_channel_message:{}", "*");
    if let Ok(callbacks) = get_on_message_callback(&String::from("*")) {
        tracing::debug!("callbacks.....",);
        callbacks
            .clone()
            .lock()
            .map_err(|err| anyhow::anyhow!("{}", err))?
            .iter()
            .all(|callback| {
                tracing::debug!("message callback.....",);
                callback(message.clone());
                true
            });
    }
    Ok(())
}

lazy_static::lazy_static! {

    ///
    pub(crate) static ref CALLBACKS: RwLock<HashMap<String, Arc<Mutex<Vec<Box<dyn Fn(TopicMessage) + Send + Sync>>>>>> = RwLock::new(HashMap::new());
}

fn get_on_message_callback(
    topic: &String,
) -> anyhow::Result<Arc<Mutex<Vec<Box<dyn Fn(TopicMessage) + Send + Sync>>>>> {
    if callback_exist(topic)? {
        let arc_topic = {
            let cache = CALLBACKS.read().map_err(|err| anyhow::anyhow!("{}", err))?;
            cache
                .get(&topic.to_lowercase())
                .map_or(Err(anyhow::anyhow!("没有查询到")), |v| Ok(v))?
                .clone()
        };
        Ok(arc_topic.clone())
    } else {
        Err(anyhow::anyhow!(""))
    }
}

pub fn set_on_message_callback(
    topic: &String,
    callback: Box<dyn Fn(TopicMessage) + Send + Sync>,
) -> anyhow::Result<()> {
    if callback_exist(topic)? {
        let arc_topic = {
            let cache = CALLBACKS.read().map_err(|err| anyhow::anyhow!("{}", err))?;
            cache
                .get(&topic.to_lowercase())
                .map_or(Err(anyhow::anyhow!("没有查询到")), |v| Ok(v))?
                .clone()
        };
        arc_topic
            .lock()
            .map_err(|err| anyhow::anyhow!("{}", err))?
            .push(callback);
        Ok(())
    } else {
        let cache_result = CALLBACKS
            .write()
            .map_err(|err| anyhow::anyhow!("获取锁失败:{}", err));

        match cache_result {
            Ok(mut cache) => {
                cache.insert(topic.to_lowercase(), Arc::new(Mutex::new(vec![callback])));
                tracing::debug!("set_on_message_callback成功!");
                Ok(())
            }
            Err(err) => {
                tracing::debug!("set_on_message_callback失敗!{}", err);
                Ok(())
            }
        }
    }
}

pub fn callback_exist(topic: &String) -> anyhow::Result<bool> {
    let exist = {
        let cache = CALLBACKS.read().map_err(|err| anyhow::anyhow!("{}", err))?;
        cache.contains_key(&topic.to_lowercase())
    };
    Ok(exist)
}

pub fn remove_on_message_callback(topic: &String) -> anyhow::Result<()> {
    if callback_exist(topic)? {
        //
        let cache_result = CALLBACKS
            .write()
            .map_err(|err| anyhow::anyhow!("获取锁失败:{}", err));

        match cache_result {
            Ok(mut cache) => {
                cache.remove(&topic.to_lowercase());
                tracing::debug!("移除message_callback成功!");
                Ok(())
            }
            Err(err) => {
                tracing::debug!("移除message_callback失敗!{}", err);
                Ok(())
            }
        }
    } else {
        Ok(())
    }
}
